package com.shujia.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo14Agg {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("Demo14Agg")


    val sc = new SparkContext(conf)

    val linesRDD: RDD[String] = sc.textFile("data/words.txt")

    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    //将rdd转换成kv格式
    val kvRDD: RDD[(String, Int)] = wordsRDD.map(word => (word, 1))

    /**
      * reduceByKey: 会再map进行预聚合，  聚合函数会应用中map端和reduce端（聚合函数会应用在分区内的聚合和分区间的聚合）
      *
      */
    val reduceRDD: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)
    reduceRDD.foreach(println)

    /**
      * aggregateByKey: 按照key进行聚合，需要两个聚合函数，一个是map端的聚合，一个是reduce端的聚合
      *
      */

    val aggRDD: RDD[(String, Int)] = kvRDD.aggregateByKey(0)(
      (u: Int, i: Int) => u + i, //分区内的聚合函数（map端的就和函数）
      (u1: Int, u2: Int) => u1 + u2 // 分区间的聚合函数（reduce的聚合函数）
    )

    aggRDD.foreach(println)

  }

}
